package com.ben.utils.ext

import com.ben.utils.flow.XSharedFlow
import com.ben.utils.flow.XStateFlow
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.stateIn

/**
 * Author: bin.yang
 * Maintainer: bin.yang
 * Date: 2021/10/11
 * Copyright: 2021 Inc. All rights reserved.
 * Desc:
 */
fun <T> CoroutineScope.collectWithEnd(
    flow: Flow<T>,
    func: suspend CoroutineScope.(T) -> Unit,
    predicate: suspend (T) -> Boolean
): Job {
    return launchIO {
        flow.first { result ->
            predicate.invoke(result).let {
                if (it) func.invoke(this, result)
                it
            }
        }
    }
}

fun <T> Flow<T>.collectWithEnd(
    scope: CoroutineScope,
    func: suspend CoroutineScope.(T) -> Unit,
    predicate: suspend (T) -> Boolean
): Job {
    return scope.launchIO {
        first { result ->
            predicate.invoke(result).let {
                if (it) func.invoke(this, result)
                it
            }
        }
    }
}

fun <T> CoroutineScope.collectOnce(flow: Flow<T>, func: suspend CoroutineScope.(T) -> Unit): Job {
    return launchIO {
        flow.first {
            func.invoke(this, it)
            true
        }
    }
}

fun <T> CoroutineScope.collectOnce(
    flow: XSharedFlow<T>,
    func: suspend CoroutineScope.(T) -> Unit
): Job {
    return launchIO {
        flow.first {
            func.invoke(this, it)
            true
        }
    }
}

fun <T> CoroutineScope.collect(flow: Flow<T>, func: suspend CoroutineScope.(T) -> Unit): Job {
    return launchIO {
        flow.collectLatest {
            func.invoke(this, it)
        }
    }
}

fun <T> Flow<T>.collect(scope: CoroutineScope, func: suspend CoroutineScope.(T) -> Unit): Job {
    return scope.launchIO {
        collectLatest {
            func.invoke(this, it)
        }
    }
}

fun <T> XSharedFlow<T>.collect(
    scope: CoroutineScope,
    func: suspend CoroutineScope.(T) -> Unit
): Job {
    return scope.launchIO {
        collectLatest {
            func.invoke(this, it)
        }
    }
}

fun <T> CoroutineScope.collect(flow: XStateFlow<T>, func: suspend CoroutineScope.(T) -> Unit): Job {
    return launchIO {
        flow.collectLatest {
            func.invoke(this, it)
        }
    }
}

fun <T> CoroutineScope.collect(
    flow: XSharedFlow<T>,
    func: suspend CoroutineScope.(T) -> Unit
): Job {
    return launchIO {
        flow.collectLatest {
            func.invoke(this, it)
        }
    }
}

fun <T> Flow<T>.toStateFlow(scope: CoroutineScope, default: T) =
    stateIn(scope, SharingStarted.WhileSubscribed(5000), default)

fun <T> Flow<T>.toXStateFlow(scope: CoroutineScope, default: T) =
    XStateFlow(stateIn(scope, SharingStarted.WhileSubscribed(5000), default))

fun <T> SharedFlow<T>.toXSharedFlow() = XSharedFlow(this)

fun <T> CoroutineScope.createMutableStateFlow(
    default: T,
    func: suspend CoroutineScope.() -> Flow<T>
): MutableStateFlow<T> {
    return MutableStateFlow(default).apply {
        launchIO {
            func.invoke(this).collect {
                tryEmit(it)
            }
        }
    }
}

fun <T> CoroutineScope.createStateFlow(
    default: T,
    func: suspend CoroutineScope.() -> Flow<T>
): StateFlow<T> {
    return createMutableStateFlow(default, func).toStateFlow(this, default)
}

fun <T> CoroutineScope.createSharedFlow(
    distinct: Boolean = true,
    func: suspend CoroutineScope.() -> Flow<T>
): XSharedFlow<T> {
    return XSharedFlow(
        MutableSharedFlow<T>(1).apply {
            launchIO {
                func.invoke(this).collect {
                    emit(it)
                }
            }
        }
    ).apply {
        if (distinct) distinctUntilChanged()
    }
}
